package org.jeecg.modules.rental.kafka;

import cn.hutool.core.util.ObjectUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Constants;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import org.apache.ibatis.annotations.Param;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jeecg.boot.starter.rabbitmq.client.RabbitMqClient;
import org.jeecg.common.base.BaseMap;
import org.jeecg.jt808.JTT808Decoder;
import org.jeecg.jt808.modbus.MqttMessageJson;
import org.jeecg.jt808.req.LocationNewMsg;
import org.jeecg.modules.app.constant.AppConstant;
import org.jeecg.modules.app.entity.AppPushRecord;
import org.jeecg.modules.app.vo.PushTargetVO;
import org.jeecg.modules.iot.model.Location;
import org.jeecg.modules.iot.model.power.TaiPower;
import org.jeecg.modules.rental.entity.RentalAccount;
import org.jeecg.modules.rental.entity.RentalEBike;
import org.jeecg.modules.rental.entity.RentalEbikeShare;
import org.jeecg.modules.rental.entity.RentalFence;
import org.jeecg.modules.rental.mapper.RentalAccountMapper;
import org.jeecg.modules.rental.mapper.RentalEBikeMapper;
import org.jeecg.modules.rental.mapper.RentalEbikeShareMapper;
import org.jeecg.modules.rental.mapper.RentalFenceMapper;
import org.jeecg.modules.rental.util.FenceMapperUtils;
import org.jeecg.modules.rental.util.FenceUtils;
import org.jeecg.modules.rental.vo.Circle;
import org.jeecg.modules.rental.vo.LocationPoint;
import org.jeecg.modules.rental.vo.Polygon;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.stereotype.Component;
import org.springframework.util.backoff.FixedBackOff;

import java.io.IOException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@RequiredArgsConstructor
@Component
public class LocationConsumer {
    private static final Logger log = LoggerFactory.getLogger(LocationConsumer.class);
    @Autowired
    private RabbitMqClient rabbitMqClient;

    @Autowired
    private RentalAccountMapper rentalAccountMapper;

    @Autowired
    private RentalEBikeMapper rentalEBikeMapper;

    @Autowired
    private RentalFenceMapper rentalFenceMapper;

    @Autowired
    private RentalEbikeShareMapper rentalEBikeShareMapper;

    /*@Test
    public void appMessageSend() {
        BaseMap json = new BaseMap();
        json.put("tenantId", "4");
        List<PushTargetVO> pushTargetList = new ArrayList<>();
        RentalAccount payAccount = rentalAccountMapper.findOneNoTenant(
                new LambdaQueryWrapper<RentalAccount>()
                        .eq(RentalAccount::getId, "1868856079715672065")
        );
        pushTargetList.add(new PushTargetVO(payAccount.getCid(), payAccount.getClientCode(), payAccount.getClientPlatform()));
        json.put("targetList", pushTargetList);
        AppPushRecord message = new AppPushRecord();
        message.setTitle("测试");
        message.setContent("第三次发送信息");
        message.setPayload("/pages/me/setting");
        json.put("message", message);
        rabbitMqClient.sendMessage(AppConstant.APP_PUSH_MESSAGE, json);
    }*/



    @KafkaListener(topics = "location", groupId = "${spring.kafka.consumer.group-id}", concurrency = "12")
    public void listen(ConsumerRecord<String, String> record) throws IOException, ParseException {

        try {
            // 解析位置信息

            ObjectMapper mapper = new ObjectMapper();
            JsonNode jsonNode = mapper.readTree(record.value());

            double latitude = jsonNode.get("latitude").asDouble();
            double longitude = jsonNode.get("longitude").asDouble();
            LocationPoint point=new LocationPoint(latitude,longitude);
            //判读是否超出围栏范围
            //1、获取对应车辆
            List<RentalEBike> eBikeList = rentalEBikeMapper
                    .listNoTenant(new QueryWrapper<RentalEBike>().eq("car_code", record.key()));
            for(RentalEBike item: eBikeList){
                //2、判断是否超出范围
                RentalFence rentalFence= rentalFenceMapper.selectById(item.getFenceId());
                if (rentalFence!=null){
                    Boolean isout= true;

                    if ("polygon".equals(rentalFence.getShapeType())){
                        Polygon polygon= FenceMapperUtils.createPolygonFromCoordinates(rentalFence.getArea());
                        isout= FenceUtils.isPointInPolygon(point,polygon);

                    }else if
                    ("circle".equals(rentalFence.getShapeType())){
                        Circle circle= FenceMapperUtils.createCircleFromCoordinates(rentalFence.getArea());
                        isout= FenceUtils.isPointInCircle(point,circle);
                    }

                    if (!isout){
                        //3、找到对应对的用户
                        List<RentalEbikeShare> shareList = rentalEBikeShareMapper
                                 //.listNoTenant(new QueryWrapper<RentalEbikeShare>().eq("ebike_id", "1868874184399589378"));
                                 .listNoTenant(new QueryWrapper<RentalEbikeShare>().eq("ebike_id", item.getId()));
                        if(ObjectUtil.isNotEmpty(shareList)){
                            List<Long> accountIds = shareList.stream().map(RentalEbikeShare::getAccountId).collect(Collectors.toList());
                            List<RentalAccount> accountList = rentalAccountMapper.listNoTenant(
                                    new LambdaQueryWrapper<RentalAccount>()
                                            .isNotNull(RentalAccount::getId)
                                            .in(RentalAccount::getId, accountIds));
                            for (RentalAccount rentalAccount : accountList) {

                                BaseMap json = new BaseMap();
                                json.put("tenantId", "4");
                                List<PushTargetVO> pushTargetList = new ArrayList<>();
                                pushTargetList.add(new PushTargetVO(rentalAccount.getCid(), rentalAccount.getClientCode(), rentalAccount.getClientPlatform()));
                                json.put("targetList", pushTargetList);
                                AppPushRecord message = new AppPushRecord();
                                message.setTitle("exceeds_fence.title");
                                message.setContent("exceeds_fence.content");
                                message.setPayload("/pages/me/setting");
                                json.put("message", message);
                                rabbitMqClient.sendMessage(AppConstant.APP_PUSH_MESSAGE, json);

                            }
                        }
                    }



                }


            }

        } catch (Exception e) {

            log.error("解析或处理位置信息失败，record: {}", record, e);
            throw e;
        }
    }

    @Bean
    public SeekToCurrentErrorHandler errorHandler(KafkaOperations<Object, Object> template) {
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
        return new SeekToCurrentErrorHandler(recoverer, new FixedBackOff(0L, 0L)); // Retry 3 times with 1-second intervals
        // ... 错误处理，失败一次的数据直接发送到死信 ...
    }


}
